package edu.purdue.cs.aggr;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Random;

import edu.purdue.cs.aggr.AccumulatorFactory.AccumulatorType;
import edu.purdue.cs.aggr.accumulators.GrepAccumulator;
import edu.purdue.cs.aggr.accumulators.MicrobenchAccumulator;
import edu.purdue.cs.aggr.accumulators.SortAccumulator;
import edu.purdue.cs.aggr.accumulators.TopKAccumulator;
import edu.purdue.cs.aggr.accumulators.WordCountAccumulator;
import edu.purdue.cs.aggr.accumulators.YAccumulator;
import edu.purdue.cs.aggr.message.EventNotificationMessage;
import edu.purdue.cs.aggr.message.EventNotificationMessage.Event;
import edu.purdue.cs.aggr.message.EventMessageBuilder;

public class InputReader {

	public final String INPUT_FILE_NAME = "/input/alldata/yahoo/yahoodata_temp";
	
	public EventNotificationMessage readInputAndBuildAccumulator(/*String hdfsNamenode, String file,*/ AccumulatorType accumulatorType, Integer accumulatorSubtype, Vertex vertex) throws AggrException {
		
		// TODO read input from HDFS

		long processingTime = -1;
		Accumulator acc = null;
		if (accumulatorType == AccumulatorType.TOPK) {
//			System.out.println("Accumulator type: TOPK");
			//For now returning a TopK accumulator
			Map<String, Object> params = new HashMap<String, Object>();
			params.put(AccumulatorFactory.PARAMETER_KVAUE, new Integer(100000)); 
			acc = AccumulatorFactory.getEmptyAccumulator(AccumulatorType.TOPK, params, vertex.getId());
			populateAccumulatorFromRandomInput(acc);
			//processingTime = puopulateAccumulatorFromFileInput(acc);

		} else  if (accumulatorType == AccumulatorType.SORT) {
//			System.out.println("Accumulator type: SORT");
			Map<String, Object> params = new HashMap<String, Object>();
			acc = AccumulatorFactory.getEmptyAccumulator(AccumulatorType.SORT, params, vertex.getId());
			
			populateAccumulatorFromRandomInput(acc);
			//processingTime = puopulateAccumulatorFromFileInput(acc);

		} else  if (accumulatorType == AccumulatorType.GREP) {
//			System.out.println("Accumulator type: GREP");
			Map<String, Object> params = new HashMap<String, Object>();
			acc = AccumulatorFactory.getEmptyAccumulator(AccumulatorType.GREP, params, vertex.getId());
			populateAccumulatorFromRandomInput(acc);
			//processingTime = puopulateAccumulatorFromFileInput(acc);

		} else  if (accumulatorType == AccumulatorType.YExperiment) {
//			System.out.println("Accumulator type: Y");
			Map<String, Object> params = new HashMap<String, Object>();
			acc = AccumulatorFactory.getEmptyAccumulator(AccumulatorType.YExperiment, params, vertex.getId());
//			puopulateAccumulatorFromRandomInput(acc);
			processingTime = populateAccumulatorFromFileInput(acc);
			
		} else  if (accumulatorType == AccumulatorType.WORDCOUNT) {
//			System.out.println("Accumulator type: WORDCOUNT");
			Map<String, Object> params = new HashMap<String, Object>();
			acc = AccumulatorFactory.getEmptyAccumulator(AccumulatorType.WORDCOUNT, params, vertex.getId());
			
//			puopulateAccumulatorFromRandomInput(acc);
			processingTime = populateAccumulatorFromFileInput(acc);

		} else  if (accumulatorType == AccumulatorType.MICROBENCH) {
//			System.out.println("Accumulator type: MICROBENCH");
			
			long startTime = System.currentTimeMillis();
			
			Map<String, Object> params = new HashMap<String, Object>();
			//TODO: Vary this from 0 to 3
			params.put(AccumulatorFactory.PARAMETER_YVALUE, accumulatorSubtype); 
			acc = AccumulatorFactory.getEmptyAccumulator(AccumulatorType.MICROBENCH, params, vertex.getId());
			
			Random rand = new Random();
			rand.setSeed(System.currentTimeMillis());
			
			for (long x = 0; x < 100000; x++) {
				acc.addValue(rand.nextLong());
			}
			
			processingTime = System.currentTimeMillis() - startTime;

		} else {
			throw new AggrException("Unknown accumulator type");
		}
		
		if (processingTime < 0) {
			throw new AggrException("Unexpected processing time");
		}
		
		vertex.setStoredAccumulator(acc);
		//TODO send 
		return EventMessageBuilder.createEventMessage(vertex, Event.LEAF_DONE_PROCESSING, new Long(processingTime));
	}

	/**
	 * 
	 * @param acc
	 * @return time to compute
	 * @throws AggrException
	 */
	private long populateAccumulatorFromFileInput(Accumulator acc) throws AggrException {
		
		System.out.println("Starting to read file input...");
		
		try {
	      if ((acc instanceof TopKAccumulator) || (acc instanceof SortAccumulator)) {
	    	  
//	    	    "taskID" : "task_201202150600_322404_m_000026",
//	    	    "startTime" : 1330539644704,
//	    	    "finishTime" : 1330539649852,
	    	  
			BufferedReader br = new BufferedReader(new FileReader(INPUT_FILE_NAME));
			List<String> inMemoryData = new ArrayList<String>(30000000);
//			String l = null;
			int i = 0;
			int nextTarget = 10485760; // 10 MB
			int interval = 10485760;
			int MBs = 0;
			int entries = 0;
			for (String l; (l = br.readLine()) != null;) {
				inMemoryData.add(l);
				entries++;
				i+=l.getBytes().length;
				if (i >= nextTarget) {
					MBs = MBs + 10;
					System.out.println("MBs :" + MBs);
					System.out.println("Entries :" + entries);
					nextTarget = nextTarget + interval;
//					System.gc();
				}
			}
			br.close();
			System.gc();
			
			// Starting to process
			System.out.println("Starting to process input...");
			long processingStart = System.currentTimeMillis();
			
	  		List<Long> intermediateData = new ArrayList<Long>(1000000);	    	  

			long startTime = -1;
			long finishTime = -1;
			boolean newRecord = false;
			int records = 0;
//			String line = null;
//			Iterator<String> lineIter = inMemoryData.iterator();		
			for (String line : inMemoryData) {
				try {
//				  line = lineIter.next();
//				  lineIter.remove();
			      if (line.trim().startsWith("\"taskID")) {
				    newRecord = true;
				  }else if (newRecord && line.trim().startsWith("\"startTime")) {
				    if(startTime >= 0) {
				    	
//					  System.out.println("Temp Num input records:" + records);
//					  System.out.println("ERROR: startTime already set...");
					  
					  //resetting
					  startTime = -1; 
					  finishTime = -1;
					  newRecord = false;
					  continue;
					}
					startTime = Long.parseLong(line.split(":|,")[1].trim());
				  } else if (newRecord && line.trim().startsWith("\"finishTime")) {
				    if(finishTime >= 0) {
//				      System.out.println("Temp Num input records:" + records);
//				      System.out.println("ERROR: finishTime already set...");

					  //resetting
					  startTime = -1; 
					  finishTime = -1;
					  newRecord = false;
					  continue;
					}  
					finishTime = startTime = Long.parseLong(line.split(":|,")[1].trim());
				  }
			      if (newRecord && (startTime > 0) && (finishTime > 0)) {
				    if (startTime > finishTime) {
//					  System.out.println("Temp Num input records:" + records);
					  throw new AggrException("Invalid record. startTime:" + startTime + "  finishTime:" + finishTime);
					}
				    intermediateData.add(finishTime-startTime);
					records++;
				
					
//					if (records==100000) {
//						break;
//					}
					
					//resetting
					startTime = -1; 
					finishTime = -1;
					newRecord = false;
			      }
				} catch (NumberFormatException e) {
					e.printStackTrace();
				}
			}
			System.out.println("# input records:" + records);
			
			
			Iterator<Long> iter = intermediateData.iterator();
			while (iter.hasNext()) {
				acc.addValue(iter.next());
			}
			
//			System.out.println("Done processing input...");
			long processingTime = System.currentTimeMillis() - processingStart;
			System.out.println("Processing time :" + processingTime);
			return processingTime;
			
	    	  
	    	  
	    	  
	    	  
	    	  
	    	  
//  	  		List<Long> inputData = new LinkedList<Long>();
//  
//  //	    	    "taskID" : "task_201202150600_322404_m_000026",
//  //	    	    "startTime" : 1330539644704,
//  //	    	    "finishTime" : 1330539649852,
//	    	   	    	  
//
//	    	  
//	    	 BufferedReader br = new BufferedReader(new FileReader(INPUT_FILE_NAME));
//	    	 String line = null;
//	    	 long startTime = -1;
//	    	 long finishTime = -1;
//	    	 boolean newRecord = false;
//	    	 int records = 0;
//	    	 while ((line = br.readLine()) != null) {
//	    	 try {
//	    	   if (line.trim().startsWith("\"taskID")) {
//	    	     newRecord = true;
//	    	   }else if (newRecord && line.trim().startsWith("\"startTime")) {
//	    	     if(startTime >= 0) {
//	    	  				    	
//  //					  System.out.println("Temp Num input records:" + records);
//  //					  System.out.println("ERROR: startTime already set...");
//  					  
//  					  //resetting
//  					  startTime = -1; 
//  					  finishTime = -1;
//  					  newRecord = false;
//  					  continue;
//  					}
//  					startTime = Long.parseLong(line.split(":|,")[1].trim());
//  				  } else if (newRecord && line.trim().startsWith("\"finishTime")) {
//  				    if(finishTime >= 0) {
//  //				      System.out.println("Temp Num input records:" + records);
//  //				      System.out.println("ERROR: finishTime already set...");
//  
//  					  //resetting
//  					  startTime = -1; 
//  					  finishTime = -1;
//  					  newRecord = false;
//  					  continue;
//  					}  
//  					finishTime = startTime = Long.parseLong(line.split(":|,")[1].trim());
//  				  }
//  			      if (newRecord && (startTime > 0) && (finishTime > 0)) {
//  				    if (startTime > finishTime) {
//  //					  System.out.println("Temp Num input records:" + records);
//  					  throw new AggrException("Invalid record. startTime:" + startTime + "  finishTime:" + finishTime);
//  					}
//  					inputData.add(finishTime-startTime);
//  					records++;
//  				
//  					
//  //					if (records==100000) {
//  //						break;
//  //					}
//  					
//  					//resetting
//  					startTime = -1; 
//  					finishTime = -1;
//  					newRecord = false;
//  			      }
//  				} catch (NumberFormatException e) {
//  					e.printStackTrace();
//  				}
//  			}
//	    	br.close();
//	    	
//  			System.out.println("# input records:" + records);
//  			
//  			// Starting to process
//  //			System.out.println("Starting to process input...");
//  			long processingStart = System.currentTimeMillis();
//  			
//  			Iterator<Long> iter = inputData.iterator();
//  			while (iter.hasNext()) {
//  				acc.addValue(iter.next());
//  			}
//  			
//  //			System.out.println("Done processing input...");
//  			return (System.currentTimeMillis() - processingStart);
	    	  
	    	  
	    	  
	    	  
	    	  
	      } else if (acc instanceof GrepAccumulator)  {

	    	  
				BufferedReader br = new BufferedReader(new FileReader(INPUT_FILE_NAME));
				List<String> inMemoryData = new ArrayList<String>(30000000);
//				String l = null;
				int i = 0;
				int nextTarget = 10485760; // 10 MB
				int interval = 10485760;
				int MBs = 0;
				int entries = 0;
				for (String l; (l = br.readLine()) != null;) {
					inMemoryData.add(l);
					entries++;
					i+=l.getBytes().length;
					if (i >= nextTarget) {
						MBs = MBs + 10;
						System.out.println("MBs :" + MBs);
						System.out.println("Entries :" + entries);
						nextTarget = nextTarget + interval;
//						System.gc();
					}
				}
				br.close();
				System.gc();
				
				long processingStart = System.currentTimeMillis();
				boolean newRecord = false;
				int records = 0;
				for (String line : inMemoryData) {
					try {
					  if (line.trim().startsWith("\"jobID")) {
						newRecord = true;
					  } else if (newRecord && line.trim().startsWith("\"startTime")) {
	
						acc.addValue(line.trim());
						records++;					

						newRecord = false;
				      }
					} catch (NumberFormatException e) {
						e.printStackTrace();
					}
				}
				
				long processingTime = System.currentTimeMillis() - processingStart;
				System.out.println("Processing time :" + processingTime);
				System.out.println("All num input records:" + records);
				return processingTime;
	    	  
//			List<String> inputData = new LinkedList<String>();
////	    	    "taskID" : "task_201202150600_322404_m_000026",
////	    	    "startTime" : 1330539644704,
////	    	    "finishTime" : 1330539649852,
//			BufferedReader br = new BufferedReader(new FileReader(INPUT_FILE_NAME));
//			String line = null;
//			long startTime = -1;
//			long finishTime = -1;
//			boolean newRecord = false;
//			int records = 0;
//			
//			while ((line = br.readLine()) != null) {
//				try {
//				  if (line.trim().startsWith("\"jobID")) {
//					newRecord = true;
//				  } else if (newRecord && line.trim().startsWith("\"startTime")) {
//
//					inputData.add(line.trim());
//					records++;
//					
////					if (records==10000) { //TODO remove
////						break;
////					}
//					
//					//resetting
//					startTime = -1; 
//					finishTime = -1;
//					newRecord = false;
//			      }
//				} catch (NumberFormatException e) {
//					e.printStackTrace();
//				}
//			}
////			System.out.println("All num input records:" + records);
//			
//			// Starting to process
////			System.out.println("Starting to process input...");
//			long start = System.currentTimeMillis();
//			Iterator<String> iter = inputData.iterator();
//			while (iter.hasNext()) {
//				acc.addValue(iter.next());
//			}
//			long end = System.currentTimeMillis();
//			
////			System.out.println("Done processing input. processing time :" + (end - start));
////			eventSender.sendEvent(Event.LEAF_STARTING_TO_AGGREGATE, new Long(end - start));
//			return (end - start);

				
				
	      } else if (acc instanceof YAccumulator)  {

	    	  
				BufferedReader br = new BufferedReader(new FileReader(INPUT_FILE_NAME));
				List<String> inMemoryData = new ArrayList<String>(30000000);
//				String l = null;
				int i = 0;
				int nextTarget = 10485760; // 10 MB
				int interval = 10485760;
				int MBs = 0;
				int entries = 0;
				for (String l; (l = br.readLine()) != null;) {
					inMemoryData.add(l);
					entries++;
					i+=l.getBytes().length;
					if (i >= nextTarget) {
						MBs = MBs + 10;
						System.out.println("MBs :" + MBs);
						System.out.println("Entries :" + entries);
						nextTarget = nextTarget + interval;
//						System.gc();
					}
				}
				br.close();
				System.gc();
				
				long processingStart = System.currentTimeMillis();
				boolean newRecord = false;
				int records = 0;
				for (String line : inMemoryData) {
					try {
					  if (line.trim().startsWith("\"jobID")) {
						newRecord = true;
					  } else if (newRecord && line.trim().startsWith("\"startTime")) {
	
						acc.addValue(line.trim());
						records++;					

						newRecord = false;
				      }
					} catch (NumberFormatException e) {
						e.printStackTrace();
					}
				}
				
				long processingTime = System.currentTimeMillis() - processingStart;
				System.out.println("Processing time :" + processingTime);
				System.out.println("All num input records:" + records);
				return processingTime;
				
		  } else if (acc instanceof WordCountAccumulator)  {
			  
				BufferedReader br = new BufferedReader(new FileReader(INPUT_FILE_NAME));
				List<String> inMemoryData = new ArrayList<String>(30000000);
//				String l = null;
				int i = 0;
				int nextTarget = 10485760; // 10 MB
				int interval = 10485760;
				int MBs = 0;
				int entries = 0;
				for (String l; (l = br.readLine()) != null;) {
					inMemoryData.add(l);
					entries++;
					i+=l.getBytes().length;
					if (i >= nextTarget) {
						MBs = MBs + 10;
						System.out.println("MBs :" + MBs);
						System.out.println("Entries :" + entries);
						nextTarget = nextTarget + interval;
//						System.gc();
					}
				}
				br.close();
				System.gc();
				
				long processingStart = System.currentTimeMillis();
				boolean newRecord = false;
				int records = 0;
				for (String line : inMemoryData) {
					
					String[] words = line.split("\\s+");
					for (String word : words) {
						acc.addValue(word.trim());
					}
					
					records++;
					if (records==100000) {
						break;
					}
				}
				
				long processingTime = System.currentTimeMillis() - processingStart;
				System.out.println("Processing time :" + processingTime);
				System.out.println("All num input records:" + records);
				return processingTime;			
				
				
				
				
				
//				List<String> inputData = new LinkedList<String>();
//				
////	    	    "taskID" : "task_201202150600_322404_m_000026",
////	    	    "startTime" : 1330539644704,
////	    	    "finishTime" : 1330539649852,
//	    	  
//			BufferedReader br = new BufferedReader(new FileReader(INPUT_FILE_NAME));
//			String line = null;
//			long startTime = -1;
//			long finishTime = -1;
//			boolean newRecord = false;
//			int records = 0;
//			while ((line = br.readLine()) != null) {
//				records++;
//				if (records==100000) {
//					break;
//				}
//				
//				String[] words = line.split("\\s+");
//				for (String word : words) {
//					inputData.add(word.trim());
//				}
//			}
////			System.out.println("All num input records:" + records);
//			
//			// Starting to process
////			System.out.println("Starting to process input...");
//			long start = System.currentTimeMillis();
//			Iterator<String> iter = inputData.iterator();
//			while (iter.hasNext()) {
//				acc.addValue(iter.next());
//			}
//			long end = System.currentTimeMillis();
//			
//			System.out.println("Done processing input...");
//			return (end - start);
		  } else {
		    throw new AggrException("Unknown accumulator type...");
		  }
			
			
			
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		return -1;
	}

	public static long populateAccumulatorFromRandomInput(Accumulator acc) throws AggrException {
		long start = System.currentTimeMillis();
		Random r = new Random(System.currentTimeMillis());
		for (int i=0; i<100000; i++) {
			if ((acc instanceof TopKAccumulator) || (acc instanceof SortAccumulator) || (acc instanceof MicrobenchAccumulator)) {
				acc.addValue(r.nextLong());
			} else if ((acc instanceof GrepAccumulator)) {
				acc.addValue(new Long(r.nextLong()).toString());
			} else {
				throw new AggrException("Unknwon accumulator");
			}
		}
		long end = System.currentTimeMillis();
		return end - start;
	}
	
}
